﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;


namespace 使用及对比
{
    /// <summary>
    /// 
    /// </summary>
    public class 接入_解析_处理_入库四步骤测试
    {


        public void Run()
        {
            //模拟接收数据
            ReceiveBuffer();
            //模拟解析与日志记录
            ResolveRecordLog();
            //模拟逻辑处理
            Process();
            //模拟将数据转换为持久化对像
            DurableQueue();
            //模拟最终入库
            WriteDurable();

            //打印记数
            UpdateSpeed();

        }
        /// <summary>
        /// 更新速度
        /// </summary>
        public void UpdateSpeed()
        {
            List<BlockQueueDemo<byte[]>> abc = new List<BlockQueueDemo<byte[]>>() {
                         q1接入,q2日志,q2解析,q3逻辑处理,q4入库持久化
                    };
            System.Threading.Tasks.Task.Factory.StartNew(() =>
            {
                while (true)
                {
                    Console.WriteLine(
                        string.Join(";\r\n", (from p in abc
                                              let sp1 = p.CreSp
                                              let sp2 = p.PopSp
                                          select    new {Name=p.Name, n1= sp1.per,sp=sp1.sp,
                                          n2 = sp2.per,
                                              sp2 = sp2.sp,
                                              ss= p.queue.Count
                                          }
                                             ).Select(p=>p.Name+"包含："+p.ss +" " +
                        "生产："+p.n1 +"^"+p.sp.ToString("F2")+
                        "/s,消费： " + p.n2 + "^" + p.sp2.ToString("F2")+"/s"
                        ))
                        );

                    System.Threading.Thread.Sleep(5000);

                }
            });


        }



        #region 接入
        BlockQueueDemo<byte[]> q1接入 = new BlockQueueDemo<byte[]>() { Name = "q1接入" };
        //模拟拼收数据服务端线程
        public void ReceiveBuffer()
        {
            System.Threading.Tasks.Task.Factory.StartNew(() =>
            {
                while (true)
                {
                    q1接入.Add(new byte[1024 * 8]);
                }
            });
        }
        #endregion

        #region 日志记录/解析线程
        BlockQueueDemo<byte[]> q2日志 = new BlockQueueDemo<byte[]>() { Name = "q2日志" };
        BlockQueueDemo<byte[]> q2解析 = new BlockQueueDemo<byte[]>() { Name = "q2解析" };

        public void ResolveRecordLog()
        {
            System.Threading.Tasks.Task.Factory.StartNew(() =>
            {
                while (true)
                {
                    byte[] result = q1接入.Pop();
                    //添加到解析与日志队列中
                    q2日志.Add(result);

                    q2解析.Add(result);
                    //模拟日志远程发送,会出现有趣的速度衰减现象
                    q2日志.Pop();
                }
            });
        }

        #endregion


        #region 逻辑处理
        BlockQueueDemo<byte[]> q3逻辑处理 = new BlockQueueDemo<byte[]>() { Name = "q3逻辑处理" };

        public void Process()
        {
            System.Threading.Tasks.Task.Factory.StartNew(() =>
            {
                while (true)
                {
                    byte[] result = q2解析.Pop();
                    //添加到逻辑处理队列中
                    q3逻辑处理.Add(result);
                }
            });
        }
        #endregion

        #region 入库持久化
        BlockQueueDemo<byte[]> q4入库持久化 = new BlockQueueDemo<byte[]>() { Name = "q4入库持久化" };

        public void DurableQueue()
        {
            System.Threading.Tasks.Task.Factory.StartNew(() =>
            {
                while (true)
                {
                    byte[] result = q3逻辑处理.Pop();
                    //添加到逻辑处理队列中
                    q4入库持久化.Add(result);
                }
            });
        }

        /// <summary>
        /// 持久化
        /// </summary>
        public void WriteDurable()
        {
            System.Threading.Tasks.Task.Factory.StartNew(() =>
            {
                while (true)
                {
                    byte[] result = q4入库持久化.Pop();
                }
            });
        }



        #endregion


    }


    /// <summary>
    /// Blocking 队列 
    /// </summary>
    public class QueueBlocking<T>
    {


        public QueueBlocking()
        {
            Quue = new BlockQueueDemo<T>();
        }

        /// <summary>
        /// 当前级队列
        /// </summary>
         public BlockQueueDemo<T> Quue { get; set; }

        /// <summary>
        /// 子级节点
        /// </summary>
        public QueueBlocking<T>  ChildrenQueue { get; set; }


        public void Start()
        {
            if (Quue != null) {
                System.Threading.Tasks.Task.Factory.StartNew(() =>
                {
                    while (true)
                    {
                        T result = Quue.Pop();
                        if (ChildrenQueue != null)
                        {
                            //循环添加到子节点中
                            ChildrenQueue.Quue.Add(result);
                        }
                    }
                });
            }

            if (ChildrenQueue != null)
            {
                ChildrenQueue.Start();
            }
        }
    }


    /// <summary>
    /// 处理速度记数类
    /// </summary>
    public class Speed
    {
        /// <summary>
        /// 速度
        /// </summary>
        public double sp { get; set; }
        /// <summary>
        /// 上一次的数据信息
        /// </summary>
        public long per { get; set; }

        /// <summary>
        /// 上一次计算速度的时间
        /// </summary>
        public DateTime PerTime { get; set; }

        /// <summary>
        /// 更新速度
        /// </summary>
        /// <param name="count"></param>
        public void UpdateSpeed(long count)
        {
            long cc = count - per;
            per = count;
            double ss= (DateTime.Now - PerTime).TotalSeconds;
            PerTime = DateTime.Now;
            sp = cc / ss;
        }
    }


    public class BlockQueueDemo<T>
    {

        #region 生产速度

        //添加总数
        long inn = 0;

        private Speed cre = new Speed() { PerTime = DateTime.Now };
        /// <summary>
        /// 生产速度
        /// </summary>
        public Speed CreSp {
            get {
                cre.UpdateSpeed(inn);
                return cre;
            }
        }


        #endregion

        #region 消费速度

        //添加总数
        long onn = 0;
        private Speed pop = new Speed() { PerTime = DateTime.Now };
        /// <summary>
        /// 消费速度
        /// </summary>
        public Speed PopSp
        {
            get
            {
                pop.UpdateSpeed(onn);
                return cre;
            }
        }

        #endregion


        /// <summary>
        /// 默认最大支持1万个对象缓存
        /// </summary>
     public   System.Collections.Concurrent.BlockingCollection<T> queue = new System.Collections.Concurrent.BlockingCollection<T>(10 * 10000);
        /// <summary>
        /// 名称
        /// </summary>
        public string Name { get; set; }
        /// <summary>
        /// 添加信息，如果队像超过最大值将阻塞线程直到信息取出为止
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="t"></param>
        public void Add(T t)
        {
            queue.Add(t);
            System.Threading.Interlocked.Increment(ref inn);
        }
        /// <summary>
        /// 取一个对像
        /// </summary>
        /// <returns></returns>
        public T Pop()
        {
            try
            {
                return queue.Take();
            }
            finally
            {
                System.Threading.Interlocked.Increment(ref onn);
            }


        }
    }
}
